package com.headfirst.dmp.tools

import com.headfirst.dmp.beans.LogBean
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * 使用自定义类LogBean，来实现数据的清洗，将数据转换成Parquet文件
  *
  * 注意：LogBean使用KryoSerializer序列化的时候，需要先注册
  *
  */
object Bzip2ParquetV2 {


  def main(args: Array[String]): Unit = {

    //1.判断参数个数
    if (args.length != 3) {
      print(
        """
          |com.hedfirst.dmp.tools.Bzip2ParquetV2
          |参数：
          |    logInputPath
          |    ResultOutputPath
          |    compressionCodec  <snappy, gzip, lzo>
        """.stripMargin)
      sys.exit(-1) //非正常退出
    }

    //2.创建session对象(使用SparkContext来创建)
    val conf = new SparkConf()
    conf.setMaster("local[*]")
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.set("spark.sql.parquet.compression.codec", args(2)) //设置压缩编码格式格式(snappy, gzip, lzo, 在Spark2.x后已经都默认snappy了，1.6等版本还是gzip)
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //设置序列化方式采用KryoSerializer方式（默认的是java序列化）
    conf.registerKryoClasses(Array(classOf[LogBean])) //注册自定义类的序列化方式

    val session: SparkSession = SparkSession.builder().config(conf).getOrCreate()

    //3.读取文件
    val line: RDD[String] = session.sparkContext.textFile(args(0))

    /**
      * 4.进行ETL处理
      */
    //过滤掉不符合条件的字段
    val arrRDD: RDD[Array[String]] = line.map(_.split(",",-1)).filter(_.length>=85)
    //将数据转换成自定义的Bean
    val logBeanRDD: RDD[LogBean] = arrRDD.map(x => LogBean(x))

    // 一种方式是传入RDD[Row]和sechma
    // 此处采用方式是传入继承了Product的自定义Bean
    val frame: DataFrame = session.createDataFrame(logBeanRDD)


    //5.讲数据写出成parquet文件
    //partitionBy可以实现按列进行分区存储
    //最终呈现的样式就是，provincename文件夹，下面有cityname的文件夹，cityname文件夹中才存放的parquet文件
    frame.write.partitionBy("provincename","cityname").parquet(args(1))

    //关闭流
    session.stop()

  }


}
